package com.teamsun.mqtt.handle;

import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import com.teamsun.mqtt.entity.vo.UserVo;
import com.teamsun.mqtt.service.ReSendService;
import com.teamsun.mqtt.service.UserPwdValidate;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttConnAckMessage;
import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttConnectPayload;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttQoS;

/**
 * 
 * @author 处理连接handle
 * 
 */
@Sharable
public class ConnectionService extends ChannelInboundHandlerAdapter {

	Logger logger = Logger.getLogger(getClass());

	/**
	 * 用于根据登录的客户端标识找channel
	 */
	ConcurrentHashMap<String, Channel> str2channel;
	/**
	 * 用于根channel 找登录的客户端
	 */
	ConcurrentHashMap<Channel, String> channel2str;
	
	

	/**用户验证
	 * 
	 */
	@Autowired
	UserPwdValidate validate;
	
	@Autowired
	ReSendService reSendService;


	public ConnectionService() {}
	
	public void channelReadComplete(ChannelHandlerContext ctx) {
		ctx.flush();
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		loginout(ctx);
		super.channelInactive(ctx);

	}

	/**
	 * 读取客户端的消息 实际的业务开始 当前仅仅支持MQTT协议，如果不是MQTT就关闭连接
	 */
	public void channelRead(ChannelHandlerContext ctx, Object msg) {

		
		if (msg instanceof MqttMessage) {

			MqttMessage message = (MqttMessage) msg;
			MqttFixedHeader fixedHeader=message.fixedHeader();
			MqttMessageType messageType = fixedHeader.messageType();
			
			switch (messageType) {
			case CONNECT:
				ack(ctx, (MqttConnectMessage) message);
				break;
			case PINGREQ:
				pong(ctx);
				break;

			case DISCONNECT:
				loginout(ctx);
				break;
			default:
				ctx.fireChannelRead(msg);
				break;
			}
		} else
			ctx.channel().close();
	}

	/**
	 * 验证用户连接合法性
	 * 
	 * 首先检查用户是否登录过了，如果登录了就拒绝重登。 检查当前需要登录验证的链接的数量，如果过多就缓存到队列 队列满了去数据库一次执行
	 * 如果比较少就一个一个验证
	 * 
	 * @param ctx
	 * @param connectMessage
	 */
	private void ack(
			final ChannelHandlerContext ctx,
			final MqttConnectMessage connectMessage) {

		MqttConnAckVariableHeader connectVariableHeader = null;
		MqttConnectReturnCode returnCode = null;
		
		
		MqttConnectPayload connectPayload = connectMessage.payload();

		String ident = connectPayload.clientIdentifier();
		Channel channel=ctx.channel();

		MqttFixedHeader fixedHeader = new MqttFixedHeader(
				MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0);

        
		if (str2channel.containsKey(ident)) {// 重复登录        
			
			Channel channelold=str2channel.get(ident);
			str2channel.remove(ident);
			channel2str.remove(channelold);
			channelold.close();
			logger.debug("重复登录关闭以前的channel"+channelold);
		}
		

		{
			
			if((returnCode=validate.validateUserLogin(new UserVo(
					connectPayload.userName(), connectPayload.password())))
					==MqttConnectReturnCode.CONNECTION_ACCEPTED){
				str2channel.put(ident, channel);
				channel2str.put(channel, ident);
				logger.debug(ident+"登陆成功");
			}
			

		}

		if (returnCode != null) {
			connectVariableHeader = new MqttConnAckVariableHeader(returnCode,
					false);
			MqttConnAckMessage mqttConnAckMessage = new MqttConnAckMessage(
					fixedHeader, connectVariableHeader);
			
			channel.write(mqttConnAckMessage);
		}
		
		
	
	}

	/**
	 * 处理心跳ping
	 * 
	 * @param ctx
	 */
	private void pong(ChannelHandlerContext ctx) {

		MqttFixedHeader fixedHeader = new MqttFixedHeader(
				MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false,
				0);

		MqttMessage mqttMessage = new MqttMessage(fixedHeader);
		ctx.channel().writeAndFlush(mqttMessage);
	}

	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

		logger.debug("异常 "+ctx.channel());
		loginout(ctx);
	}

	/**
	 * 合法退出
	 * 
	 * @param channel
	 */
	void loginout(ChannelHandlerContext handlerContext) {
		Channel channel = handlerContext.channel();
		String iden = channel2str.get(channel);

		if (iden != null) {
			str2channel.remove(iden);
			reSendService.remmoveFromClientId(iden,false);
			logger.info(iden + "退出,在线人数\t" + str2channel.size());

			channel2str.remove(channel);
			logger.info(channel + "断开" + channel2str.size());
		}

		handlerContext.close();
	}

	public void setStr2channel(ConcurrentHashMap<String, Channel> str2channel) {
		this.str2channel = str2channel;
	}

	public void setChannel2str(ConcurrentHashMap<Channel, String> channel2str) {
		this.channel2str = channel2str;
	}

	public ConcurrentHashMap<String, Channel> getStr2channel() {
		return str2channel;
	}

	public ConcurrentHashMap<Channel, String> getChannel2str() {
		return channel2str;
	}



	
	
}
